package com.zx.learn.flink.source.custom;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Random;
import java.util.UUID;

/**
 * 自定义数据源，每秒钟生成一个订单信息
 */
@Slf4j
class MySelfDefineSource extends RichParallelSourceFunction<OrderBean> {
    //定义是否循环生成数据的标记
    private boolean isRunning = true;

    /**
     * 一个线程只会被执行一次
     * 一般用来初始化资源，比如创建mysql的连接
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        log.debug("open方法被初始化...");
    }

    /**
     * 释放资源
     *
     * @throws Exception
     */
    @Override
    public void close() throws Exception {
        super.close();
        System.out.println("close方法被释放...");
    }

    /**
     * 核心方法，生成数据
     */
    @Override
    public void run(SourceContext<OrderBean> ctx) throws Exception {
        int loop = 0;
        Random random = new Random();
        while (isRunning) {
            //订单id
            String orderId = UUID.randomUUID().toString();
            //用户id
            String userId = String.valueOf(random.nextInt(3));
            //订单金额
            int money = random.nextInt(1000);
            //时间
            long time = System.currentTimeMillis();

            //ctx返回数据
            ctx.collect(new OrderBean(orderId, userId, money, time));

            //设置休眠时间
            Thread.sleep(1000L);
            if (loop++ > 2) {
                break;
            }
        }
    }

    /**
     * 取消方法
     */
    @Override
    public void cancel() {
        isRunning = false;
    }
}
